﻿using Confluent.Kafka;
using LJ.EventBus.FastDog.Data;
using LJ.EventBus.FastDog.Transport.Abstractions;
using LJ.EventBus.FastDog.Transport.Abstractions.Exceptions;
using Microsoft.Extensions.Logging;
using System.Text;

namespace LJ.EventBus.FastDog.Transport.Kafka.Internal;

public class KafkaTransport(ILogger<KafkaTransport> logger, IConnectionPool connectionPool) : ITransport
{
    private readonly IConnectionPool _connectionPool = connectionPool;
    private readonly ILogger _logger = logger;

    public async Task SendAsync(TransportMessage message, CancellationToken cancellationToken = default)
    {
        var producer = _connectionPool.RentProducer();

        try
        {
            var headers = new Headers();

            foreach (var header in message.Headers)
            {
                headers.Add(header.Value != null
                    ? new Header(header.Key, Encoding.UTF8.GetBytes(header.Value))
                    : new Header(header.Key, null));
            }

            var result = await producer.ProduceAsync(message.GetName(), new Message<string, byte[]>
            {
                Headers = headers,
                Key = message.Headers.TryGetValue(KafkaHeaders.KafkaKey, out var kafkaMessageKey) &&
                      !string.IsNullOrEmpty(kafkaMessageKey)
                    ? kafkaMessageKey
                    : message.GetId(),
                Value = message.Body.ToArray()!
            }, cancellationToken);

            if (result.Status is PersistenceStatus.Persisted or PersistenceStatus.PossiblyPersisted)
            {
                _logger.LogDebug("kafka topic message [{message}] has been published.", message);

                return;
            }

            throw new PublisherSentFailedException("kafka message persisted failed!");
        }
        catch (PublisherSentFailedException)
        {
            throw;
        }
        catch (Exception ex)
        {
            throw new PublisherSentFailedException(ex.Message, ex);
        }
        finally
        {
            _connectionPool.Return(producer);
        }
    }
}
